package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseApp;
import com.atguigu.gmall.realtime.bean.TrafficPageViewBean;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.function.DorisMapFunction;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.DorisUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/8/8 10:44
 */
public class Dws_02_DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new Dws_02_DwsTrafficVcChArIsNewPageViewWindow().start(
            40003,
            2,
            "Dws_02_DwsTrafficVcChArIsNewPageViewWindow",
            GmallConstant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       DataStreamSource<String> stream) {
        // 1. 把json 字符串解析出来, 封装到一个对象中
        SingleOutputStreamOperator<TrafficPageViewBean> beanStream = parseToPojo(stream);
        // 2. 开窗聚合
        SingleOutputStreamOperator<TrafficPageViewBean> resultStream = windowAndAgg(beanStream);
        
        // 3. 写出到 doris
        writeToDoris(resultStream);
    }
    
    private void writeToDoris(SingleOutputStreamOperator<TrafficPageViewBean> resultStream) {
        resultStream
            .map(new DorisMapFunction<>())
            .sinkTo(DorisUtil.getDorisSink("gmall2023.dws_traffic_vc_ch_ar_is_new_page_view_window"));
    }
    
    private SingleOutputStreamOperator<TrafficPageViewBean> windowAndAgg(SingleOutputStreamOperator<TrafficPageViewBean> beanStream) {
        /*
        简单
            sum
            max maxBy
            min minBy
        
        复杂
            reduce
                一般是输入类型和输出类型一致
            aggregate
                 可以不一致
            process
                自己完成
         */
        
       return beanStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((bean, ts) -> bean.getTs())
                    .withIdleness(Duration.ofSeconds(60))  // 解决数据倾斜导致的水印不更新
            )
            .keyBy(bean -> bean.getVc() + "_" + bean.getCh() + "_" + bean.getAr() + bean.getIsNew())
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            
            .reduce(
                new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1,
                                                      TrafficPageViewBean value2) throws Exception {
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
    
                        return value1;
                    }
                },
                new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                    @Override
                    public void process(String key,  // key的值
                                        Context ctx, // 上下文
                                        Iterable<TrafficPageViewBean> elements, // 有且仅有一个值: 前面聚合的最终结果
                                        Collector<TrafficPageViewBean> out) throws Exception {
                        TrafficPageViewBean bean = elements.iterator().next();
                        bean.setStt(DateFormatUtil.tsToDateTime(ctx.window().getStart()));
                        bean.setEdt(DateFormatUtil.tsToDateTime(ctx.window().getEnd()));
                        bean.setCur_date(DateFormatUtil.tsToDate(ctx.window().getEnd()));  // 日期
                        out.collect(bean);
    
                    }
                }
            );
    }
    
    private SingleOutputStreamOperator<TrafficPageViewBean> parseToPojo(DataStreamSource<String> stream) {
        return stream
            .map(JSON::parseObject)
            .keyBy(obj -> obj.getJSONObject("common").getString("mid"))
            .map(new RichMapFunction<JSONObject, TrafficPageViewBean>() {
                
                private ValueState<String> lastVisitDateState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    lastVisitDateState = getRuntimeContext()
                        .getState(new ValueStateDescriptor<String>("lastVisitDate", String.class));
                }
                
                @Override
                public TrafficPageViewBean map(JSONObject obj) throws Exception {
                    JSONObject common = obj.getJSONObject("common");
                    String vc = common.getString("vc");
                    String ar = common.getString("ar");
                    String ch = common.getString("ch");
                    String isNew = common.getString("is_new");
                    
                    JSONObject page = obj.getJSONObject("page");
                    Long durSum = page.getLong("during_time");
                    
                    Long ts = obj.getLong("ts");
                    String today = DateFormatUtil.tsToDate(ts);
                    
                    long uvCt = 0;
                    // 如果是该设置的一条置为 1  TODO
                    if (!today.equals(lastVisitDateState.value())) {
                        uvCt = 1L;
                        // 更新状态
                        lastVisitDateState.update(today);
                    }
                    
                    
                    return new TrafficPageViewBean(
                        "", "",
                        vc, ch, ar, isNew,
                        "", // 当天日期, 等到聚合完成之后再补充
                        uvCt, 0L, 1L, durSum,
                        ts,
                        common.getString("sid")
                    );
                }
            })
            .keyBy(TrafficPageViewBean::getSid)
            .process(new KeyedProcessFunction<String, TrafficPageViewBean, TrafficPageViewBean>() {
                
                private ValueState<Boolean> isFirstState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    isFirstState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("isFirst", Boolean.class));
                }
                
                @Override
                public void processElement(TrafficPageViewBean value,
                                           Context ctx,
                                           Collector<TrafficPageViewBean> out) throws Exception {
                    if (isFirstState.value() == null) {
                        value.setSvCt(1L);
                        isFirstState.update(false);
                    }
                    out.collect(value);
                }
            });
    }
}
/*
流量域版本-渠道-地区-访客类别粒度页面浏览各窗口汇总表

    会话数  page_log
        1. 新会话
            last_page_id = null 表示开启了一个新会话
                有点不准
        2. 新会话
            如果有 session_id, 比较准确的找到新会话
                使用状态, 状态为空就是该会话的第一条日志, 其他的不是第一条
    
    页面浏览数（pv） page_log
        直接读取 page_log 日志, 每一条日志对 pv 的值贡献 1
    
    、浏览总时长、
        during_time 直接聚和相加
    
    独立访客数（uv） page_log
        统计当天一共有多少个不同设备 id
            找到每天每台设备的第一条页面日志, 这条日志对 uv 的贡献是 1, 改设备的其他访问记录都贡献 0
            
            如何找到当天的第一条?
                使用状态
                    状态记录下该设备的最后一次访问的日期
                    
                    如果这条数据的日期和状态中的值不相等, 则是第一条, 更新状态
                    否则就不是第一条,不用更新状态
        
------
 ....   sv    pv  uv    dur_sum
        1/0   1   1/0   10000
        
 分组keyBy: 版本-渠道-地区-访客类别
 开窗, 聚合:
    把制表相加
    
 写出到 doris
*/